package com.SparkCore.RDD.Serial

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark01_RDD_Serial {
  def main(args: Array[String]): Unit = {

    //创建连接
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Serial")
    //创建SparkContext，作为提交Spark APP的入口
    val sc: SparkContext = new SparkContext(sparkConf)
    //创建RDD算子
    val rdd = sc.makeRDD(List("hello word", "hello scala", "hello Spark", "hive"))
    //创建一个Search对象
    val search = new Search("h")
    //函数传递，打印：Error Task not serializable
    search.getMatch1(rdd).collect().foreach(println)
    //关闭连接
    sc.stop()
  }
}
//查询对象
//类的构造参数其实是类的构造属性，构造参数需要进行闭包检测，其实就等同于闭包检测
class Search(query: String) extends Serializable {
  def isMatch(s: String): Boolean = {
    s.contains(query)
  }

  // 函数序列化案例
  def getMatch1 (rdd: RDD[String]): RDD[String] = {
    //rdd.filter(this.isMatch)
    rdd.filter(isMatch)
  }
  // 属性序列化案例
  def getMatch2(rdd: RDD[String]): RDD[String] = {
    //rdd.filter(x => x.contains(this.query))
    rdd.filter(x => x.contains(query))
    //val q = query
    //rdd.filter(x => x.contains(q))
  }
}
